iT邦幫忙

2024 iThome 鐵人賽

DAY 26
0
Python

從概念到應用:Python實戰開發學習之旅系列 第 26

[Day25] Python專案 - 網頁開發 - (4) Fast API 進階後端工程師該思考的幾件事

  • 分享至 

  • xImage
  •  

目標

https://ithelp.ithome.com.tw/upload/images/20241010/201210524xN5wPyLrs.png

我們在Day 23 學習到了後端框架的選擇以及簡單的CRUD
[Day23] Python專案 - 網頁開發 - (2) 高速的服務提供者Fast API

我們在Day 24 學習到了後端與前端交互的應用
[Day24] Python專案 - 網頁開發 - (3) 前端的至尊React與Python之間的交流

今天我們回歸到後端的老本行,畢竟還是要以Python為主XDD
/images/emoticon/emoticon28.gif

今天我們要探討的是身為一位後端工程師
除了提供API服務以外
還有幾件事情要注意(比如說:我們昨天學到的cors也是需要注意的)

今天我會分享在學習後端時
我很希望前人能提醒我或交會我的幾項技術:

  1. API文件
  2. 認證機制
  3. 資料庫應用

1. API文件

在開發後端 API 時,API 文件扮演著關鍵角色,因為它不僅幫助開發者理解和使用 API,也能確保 API 使用的一致性和可維護性。

1.1 說明API文件需知道的事項

1.結構清晰且易於理解:

API 文件應該有清晰的結構,方便開發者快速找到需要的資源或方法。每個端點的功能、參數、請求和響應格式都需要詳細描述。

2.正確描述請求與響應:

說明每個 API 端點需要的 HTTP 方法(GET、POST、PUT、DELETE 等)和參數,並清楚描述響應的格式(JSON、XML 等)。必要時,提供具體的例子。

3.處理錯誤代碼與消息:

清楚地列出各種可能的 HTTP 狀態碼(如 200、400、500)及其對應的錯誤訊息。這有助於使用者快速診斷問題並進行調試。

4.授權與驗證:

如果 API 使用身份驗證(如 JWT、API Key、OAuth),文件中應詳細說明如何通過這些驗證進行安全請求,並提供授權流程的範例。

5.版本控制:

隨著 API 的迭代更新,不同版本可能會引入不相容的變更,因此應記錄 API 的版本及變更歷史,並保持過去版本的文件可用。

6.自動生成與持續更新:

使用工具(如 OpenAPI/Swagger)來自動生成 API 文件,這樣可以減少手動錯誤,並確保文件隨著代碼的更新而更新。


1.2 API 文件應用場景

1. 開發團隊內部協作:

API 文件是前後端團隊協作的橋樑,讓前端開發者在沒有後端系統完全實現的情況下,能根據 API 文件模擬請求,並進行開發。

2.第三方開發者使用:

若 API 提供給第三方開發者使用,詳細的文件可以幫助他們理解如何集成和使用你的 API,並減少溝通成本

3.自動化測試與工具集成:

基於標準化的 API 文件(如 OpenAPI),可以生成自動化測試、模擬服務器或 SDK,提升測試效率和穩定性。


1.3 FastAPI 與 OpenAPI 的關係

FastAPI 與 OpenAPI 的關係

OpenAPI 規範:OpenAPI 是一個標準規範,用於描述 RESTful API。它定義了 API 的路由、請求參數、響應數據格式、授權方式等細節,從而使得 API 可以通過統一的方式進行文檔生成、自動化測試以及客戶端代碼生成等操作。

自動生成 OpenAPI 文檔:FastAPI 自動為你編寫的 API 端點生成 OpenAPI 規範文檔。當你創建新的路由或 API 端點時,FastAPI 會根據端點的路由、請求方法、請求參數和響應模型等,動態生成符合 OpenAPI 規範的文件。

Swagger UI 與 ReDoc:FastAPI 提供了內建的 Swagger UI 和 ReDoc,用於展示和測試 API:

Swagger UI:一個互動式的網頁,允許開發者直接在瀏覽器中測試 API 請求。FastAPI 的 Swagger UI 是基於自動生成的 OpenAPI 規範構建的。
ReDoc:另一種用於展示 OpenAPI 規範的界面,提供更多的視覺化方式來展示 API 文檔。

如何查看 FastAPI 生成的 OpenAPI 文檔

大家可以把這兩天建立的customer的CRUD API重新啟動
無須安裝套件,直接在網址後面選擇路徑(docs 或是 redoc) 就可以看到文件摟

http://127.0.0.1:8000/docs   # Swagger UI

https://ithelp.ithome.com.tw/upload/images/20241010/20121052ELwpAJGJc0.png

http://127.0.0.1:8000/redoc   # ReDoc

https://ithelp.ithome.com.tw/upload/images/20241010/20121052Ggvlf5aIbT.png

OpenAPI 文檔的自定義

FastAPI 允許你在生成的 OpenAPI 文檔中進行自定義,例如:

標題:可以自定義 API 文檔的標題和版本信息。
描述:可以在文檔中添加描述,提供 API 的總覽。
標籤:可以為不同的端點分類,這樣文檔會更有組織性。

可以改程式碼這些參數讓api的自定義更有識別度喔

from fastapi import FastAPI

app = FastAPI(
    title="Roni's API",
    description="This is a sample CURD Demo with FastAPI",
    version="1.0.0"
)

使用場景
內部開發和協作:基於 OpenAPI 生成的文檔可以幫助前後端團隊協作,更加清晰地理解 API 的結構與功能。
對外開放 API:當你開發對外的 API 服務時,自動生成的 OpenAPI 文檔讓其他開發者能快速理解和使用你的 API。


2.認證機制

功能:API 認證是確保只有授權用戶能夠訪問特定資源的過程。這通常涉及使用身份驗證標頭、API 密鑰、OAuth 或 JWT(JSON Web Tokens)等技術。
安全性:良好的認證機制能防止未授權訪問,保護敏感數據。
實施:後端工程師需要設計認證流程,並確保所有敏感操作都經過適當的授權檢查。

通常認證方式又會有3種分別是Cookie + Session 認證OAuth 認證JWT 認證

認證方法 優點 缺點 適用場景
Cookie + Session 簡單易實現;自動處理會話狀態 伺服器負擔大;跨域請求管理複雜 傳統 Web 應用,需要持久的會話狀態
OAuth 不需暴露用戶憑據;靈活的授權和 Token 管理 實現較為複雜;依賴第三方認證系統 第三方應用訪問外部資源,如社交登入或應用整合
JWT 無狀態、輕量;跨服務器應用方便 Token 無法撤銷;需要注意安全性管理 RESTful API,分佈式系統,微服務架構

2.1 Cookie + Session 認證

https://ithelp.ithome.com.tw/upload/images/20241010/20121052DCkb2KXKjL.png

這是傳統的認證方式,主要用於網站應用,特別是在伺服器與用戶之間的狀態保持上。

工作原理:

用戶登入時,伺服器驗證用戶身份(例如用戶名和密碼)。
驗證成功後,伺服器會生成一個唯一的 Session ID,並將該 ID 儲存在伺服器上。
伺服器會將該 Session ID 以 Cookie 的形式存放在用戶瀏覽器中。
每次用戶請求 API 時,瀏覽器會自動將 Cookie 包含在請求中,伺服器會根據 Cookie 中的 Session ID 查找對應的用戶身份。

優點:

比較簡單,易於使用和實現。
Cookie 可以自動附帶在同一域名下的每個請求中,減少手動傳遞。

缺點:

需要在伺服器端存儲 Session 信息,當用戶規模變大時,伺服器負擔增加。
對於跨域 API 請求,處理 CORS(跨域資源共享)和 Cookie 的安全性問題較為麻煩。

2.2 OAuth 認證

https://ithelp.ithome.com.tw/upload/images/20241010/201210526G2TsTDdCG.png

OAuth(Open Authorization)是一個授權框架,通常也被稱為第三方認證,用於安全地讓第三方應用訪問用戶的資源,而不需要暴露用戶的憑據。它特別適用於允許應用程序訪問其他服務(如 Google、Facebook)上的用戶數據。

工作原理:

用戶嘗試登入第三方應用,應用將用戶重定向到 OAuth 提供者(如 Google)。
用戶在 OAuth 提供者那裡登入並授權應用訪問資源。
OAuth 提供者返回一個短期有效的授權碼。
應用使用該授權碼向 OAuth 提供者請求 Access Token。
Access Token 用於授權應用程序訪問用戶的受保護資源。

優點:

提供授權而不需要用戶直接給第三方應用自己的憑據(如密碼)。
Access Token 可以設定有效期,並可單獨撤銷授權。

缺點:

設置與實現較為複雜。
需要依賴第三方的認證系統。 (如果以後透過這些認證方式進入的網站改過規範或是倒閉!?應該不會)/images/emoticon/emoticon16.gif

會需要重新評估或重寫認證界街的應用

應用場景:

用於第三方應用整合(如允許應用訪問 Google Drive 文件)。
適合涉及第三方服務器資源的情況。

2.3 JWT(JSON Web Token)認證

https://ithelp.ithome.com.tw/upload/images/20241010/20121052Hh7MgVLuPu.png

JWT 是一種無狀態的認證方法,這意味著不需要在伺服器上存儲用戶的會話狀態。JWT 由三部分構成:標頭(header)、有效載荷(payload)和簽名(signature),並通過簽名來確保數據的完整性。

https://jwt.io/

https://ithelp.ithome.com.tw/upload/images/20241010/2012105283c0q5PlrT.png

基本結構:

JWT(JSON Web Token)是一種安全的 Token 格式,用於身份驗證和信息交換。它由三部分構成,每個部分都用「.」分隔:

標頭(Header):包含了令牌的元數據,如加密演算法的類型和類別。
有效載荷(Payload):包含實際的數據或聲明,如用戶信息、權限、Token 有效期等。
簽名(Signature):用來驗證令牌的完整性,確保內容未被篡改。

Header比較少修改內容
Payload可以是依照業務邏輯需求塞入的用戶Data(roni name 、Engineer 之類的訊息)
通常Payload也會比較長
Signature這段通常前端在送JWT Token之後,後端會拿key去驗證對對

工作原理:

用戶登錄時,伺服器驗證用戶憑據(如用戶名和密碼)。
驗證成功後,伺服器生成一個 JWT Token,其中包含了用戶信息和過期時間。
JWT Token 返回給用戶,並儲存在用戶的瀏覽器(通常是 localStorage 或 sessionStorage)。
用戶每次請求 API 時,將 JWT Token 附加在請求頭部。
伺服器收到請求後,通過驗證 JWT 的簽名來確認請求合法性,並解析其中的用戶信息。

優點:

無狀態:伺服器無需存儲會話信息,大幅減少伺服器負擔。
Token 可以跨多個服務器使用,方便在微服務架構中使用。
自包含(Self-contained):JWT 中包含了用戶的所有信息,不需要依賴伺服器查詢。

缺點:

Token 無法在發出後撤銷,除非設置很短的有效期,否則一旦被竊取可能會造成風險。
Token 在客戶端儲存時需要注意安全,避免 XSS 攻擊。

應用場景:

適合於需要無狀態認證的 RESTful API。
非常適合分佈式系統或微服務架構,因為 JWT 可以跨多個服務進行驗證。

2.4 JWT Token 與 Fast API

因為目前Web Application除了透過Oauth認證以外最常用的就是JWT Token了
從這邊我們來測試製作兩個API吧 (製作Token、驗證身分並回傳資料)!!

安裝JWT套件並把相關認證也裝完整

pip install fastapi[all] python-jose

製作Token 跟 驗證的API

https://ithelp.ithome.com.tw/upload/images/20241010/201210521V5v3u21rd.png

POST /token:

客戶端發送帳號和密碼到伺服器,伺服器驗證用戶資料後生成 JWT Token 並返回給客戶端。
客戶端保存 Token 以便後續請求使用。

GET /profile:

客戶端發送包含 Bearer Token 的請求來獲取 Profile 資料。
伺服器驗證 Token 是否正確,然後返回對應的用戶 Profile 資料。

import json
from fastapi import FastAPI, HTTPException, Depends, status, Header
from fastapi.security import OAuth2PasswordRequestForm
from jose import JWTError, jwt
from datetime import datetime, timedelta
from pydantic import BaseModel

app = FastAPI()

# 配置 JWT 秘鑰和算法
SECRET_KEY = "mysecret"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# 模擬的用戶資料庫
fake_users_db = {
    "roni": {
        "username": "roni",
        "password": "12345678",  # 簡化的密碼處理
    }
}

# 模擬的 Profile 資料
fake_profiles_db = {
    "roni": {
        "username": "roni",
        "full_name": "Roni Albahari",
        "email": "roni@example.com",
    }
}

# Token 模型
class Token(BaseModel):
    access_token: str
    token_type: str

# 生成 JWT Token
def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# 登入並生成 Token 並存儲到 token.json
@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = fake_users_db.get(form_data.username)
    if not user or user["password"] != form_data.password:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(data={"sub": user["username"]}, expires_delta=access_token_expires)

    # 將 Token 存儲到 token.json
    with open("token.json", "w") as token_file:
        json.dump({"access_token": access_token}, token_file)

    return {"access_token": access_token, "token_type": "bearer"}

# 解析 Bearer Token,並返回用戶 Profile
@app.get("/profile")
async def get_profile(Authorization: str = Header(None)):
    if not Authorization:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="No authentication provided")

    token = Authorization.split(" ")[1]
    
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None or username not in fake_profiles_db:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
        
        # 回傳 Profile 資料
        return fake_profiles_db[username]

    except JWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token validation failed")


驗證結果

送出帳密認證

http://127.0.0.1:8000/token
username roni
password 123456

https://ithelp.ithome.com.tw/upload/images/20241010/20121052x6uHxdDelq.png

送出帳密認證

http://127.0.0.1:8000/profile
選擇Bearer Token 把剛剛創造的token貼上去送出

https://ithelp.ithome.com.tw/upload/images/20241010/20121052gW79GT7Va8.png

說明創造JWT TOKEN的function

前面設定的密鑰、演算法、token過期時間

SECRET_KEY = "mysecret"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

創建token的function

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt
Step 1 輸入資料處理:
使用 OAuth2PasswordRequestForm 來處理來自用戶的登入請求,包含了用戶輸入的 username 和 password。

Step 2 檢查用戶和密碼:

fake_users_db.get(form_data.username):根據用戶輸入的帳號名稱,從模擬的用戶數據庫中查找該用戶。
如果用戶不存在或密碼不匹配,則拋出一個 401 錯誤,並返回「帳號或密碼不正確」的訊息。

Step 3設置過期時間並生成 Token:

timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES):設定 JWT Token 的過期時間,這裡設置為 30 分鐘。
create_access_token():呼叫前面定義的函數,根據用戶名生成一個 JWT Token,並且包含過期時間。

Step 4儲存 Token:

將生成的 JWT Token 寫入 token.json 文件中,以便在後續請求中可以檢索和使用。


Step 5 返回 Token:

最後,返回一個包含生成的 access_token 和 token_type(Bearer)的 JSON 給客戶端。前端接收到這個 Token 後,後續每次訪問受保護的 API 時,會將此 Token 添加到 Authorization Header 中。

Token(Post) API的說明

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = fake_users_db.get(form_data.username)
    if not user or user["password"] != form_data.password:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(data={"sub": user["username"]}, expires_delta=access_token_expires)

    # 將 Token 存儲到 token.json
    with open("token.json", "w") as token_file:
        json.dump({"access_token": access_token}, token_file)

    return {"access_token": access_token, "token_type": "bearer"}

解釋產生token的post api:



Step1 輸入資料處理:

使用 OAuth2PasswordRequestForm 來處理來自用戶的登入請求,包含了用戶輸入的 username 和 password。

Step2 檢查用戶和密碼:

fake_users_db.get(form_data.username):根據用戶輸入的帳號名稱,從模擬的用戶數據庫中查找該用戶。
如果用戶不存在或密碼不匹配,則拋出一個 401 錯誤,並返回「帳號或密碼不正確」的訊息。

Step3 設置過期時間並生成 Token:

timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES):設定 JWT Token 的過期時間,這裡設置為 30 分鐘。
create_access_token():呼叫前面定義的函數,根據用戶名生成一個 JWT Token,並且包含過期時間。

Step4 儲存 Token:

將生成的 JWT Token 寫入 token.json 文件中,以便在後續請求中可以檢索和使用。

Step5 返回 Token:

最後,返回一個包含生成的 access_token 和 token_type(Bearer)的 JSON 給客戶端。前端接收到這個 Token 後,後續每次訪問受保護的 API 時,會將此 Token 添加到 Authorization Header 中。

Profile(Get) API 說明

if not Authorization:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="No authentication provided")

    token = Authorization.split(" ")[1]

如果前端沒有帶token就直接把user踢出去了(沒有驗證還想進來XDD)

 try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None or username not in fake_profiles_db:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
        
        # 回傳 Profile 資料
        return fake_profiles_db[username]

    except JWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token validation failed")

這邊很簡單其實就是比對token
另外就是SECRET_KEY都是存到後端server我們可以透過這個去解開看看是不是合法Token
也可以到官方去試看看

https://ithelp.ithome.com.tw/upload/images/20241010/201210521tMU2nAVrl.png

把seceryKey輸入上去就會看到是否合法token
也可以看到我們token erpire日期
我的創建日期是2024/10/10 16:51
過期時間就是 2024/10/10 17:21

3.資料庫

資料庫是一種系統化的數據存儲方式,用於管理、檢索和操作數據。資料庫可以分為多種類型,包括關聯式資料庫、NoSQL 資料庫等。

常見資料庫類型

  1. 關聯式資料庫(Relational Database)

    • 使用表格來存儲數據,並透過 SQL(結構化查詢語言)進行查詢和管理。
    • 例子:MySQL、PostgreSQL、Oracle。
  2. NoSQL 資料庫

    • 不使用表格結構,適合處理大規模的非結構化數據。
    • 例子:MongoDB(文件型)、Redis(鍵值型)、Cassandra(列族型)。
  3. 圖形資料庫(Graph Database)

    • 專門用於存儲和查詢圖形結構的數據,適合處理複雜的關係。
    • 例子:Neo4j、ArangoDB。

應用場景

  1. 電子商務

    • 存儲產品信息、用戶資料和訂單記錄,支持快速查詢和交易處理。
  2. 社交媒體

    • 管理用戶賬號、帖子、評論和朋友關係,支持高效的數據檢索。
  3. 金融服務

    • 處理交易記錄、客戶信息和風險評估,確保數據安全和準確性。
  4. 內容管理系統(CMS)

    • 存儲文章、圖片和用戶生成內容,支持靈活的數據查詢和管理。
  5. 物聯網(IoT)

    • 收集和分析來自各種設備的數據,支持實時監控和決策。

關聯資料庫 vs NoSQL 資料庫

特性 關聯資料庫(Relational Database) NoSQL 資料庫(Non-Relational Database)
優點 - 支持複雜查詢和事務處理 - 高度可擴展,適合大規模數據存儲
- 嚴格的數據一致性和完整性(ACID 屬性) - 彈性的數據模型,支持多種數據類型
- 結構化數據管理,易於理解和使用 SQL 語言 - 快速讀寫操作,適合實時應用
缺點 - 擴展性有限,通常需要垂直擴展 - 數據一致性相對較低,可能不支持 ACID 屬性
- 複雜的數據模型設計和維護 - 查詢語言不統一,可能需要學習不同的查詢方式
應用場景 - 電子商務系統,需管理複雜訂單和交易 - 社交媒體平台,需處理大量非結構化用戶生成內容
- 金融服務,需確保高安全性和數據一致性 - 物聯網(IoT),需快速處理來自多個設備的大量數據
- 內容管理系統(CMS),需管理結構化內容 - 大數據分析,需處理多樣化的數據集

提供幾個粒子可以給大家思考!

User Profile或是社交貼文很適合用NoSQL
選課系統或是簽到表很適合用關聯資料庫

合體Fast API Customer 的(MongoDB)

https://www.mongodb.com/try/download/community

MongoDB是我很喜歡的NoSQL
我會用它來做DEMO 範例
安裝完後可以透過詳細程序看到是否有啟用(預設會啟用喔)

https://ithelp.ithome.com.tw/upload/images/20241010/20121052uOofR2sjTY.png

先安裝MONGODB相關的套件庫

pip install motor

TIPS PyMongo(不建議)
這個是官方提供的套件庫,我不建議。因為他是基於同步模式使用的。
也就是你的早餐店老闆娘會,一個步驟一個步驟做完才會處理完成給你。
昨天Day24提到的梗
失去了非同步的意義。

實作程式碼

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from fastapi.middleware.cors import CORSMiddleware
from motor.motor_asyncio import AsyncIOMotorClient
from bson import ObjectId

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# MongoDB 連接
MONGO_URL = "mongodb://localhost:27017"
client = AsyncIOMotorClient(MONGO_URL)
db = client.customer_db
customer_collection = db.customers

# 顧客資料模型
class Customer(BaseModel):
    name: str
    email: str
    address: str = None

class CustomerInDB(Customer):
    id: str

# 創建顧客
@app.post("/customers/", response_model=CustomerInDB)
async def create_customer(customer: Customer):
    customer_dict = customer.dict()
    result = await customer_collection.insert_one(customer_dict)
    created_customer = await customer_collection.find_one({"_id": result.inserted_id})
    return CustomerInDB(id=str(created_customer["_id"]), **created_customer)

# 獲取所有顧客
@app.get("/customers/", response_model=list[CustomerInDB])
async def get_customers():
    customers = await customer_collection.find().to_list(1000)
    return [CustomerInDB(id=str(customer["_id"]), **customer) for customer in customers]

# 更新顧客
@app.put("/customers/{customer_id}", response_model=CustomerInDB)
async def update_customer(customer_id: str, updated_customer: Customer):
    result = await customer_collection.update_one(
        {"_id": ObjectId(customer_id)},
        {"$set": updated_customer.dict()}
    )
    if result.modified_count == 0:
        raise HTTPException(status_code=404, detail="顧客未找到")
    updated_doc = await customer_collection.find_one({"_id": ObjectId(customer_id)})
    return CustomerInDB(id=str(updated_doc["_id"]), **updated_doc)

# 刪除顧客
@app.delete("/customers/{customer_id}")
async def delete_customer(customer_id: str):
    result = await customer_collection.delete_one({"_id": ObjectId(customer_id)})
    if result.deleted_count == 0:
        raise HTTPException(status_code=404, detail="顧客未找到")
    return {"message": "顧客成功刪除"}

驗證結果

https://ithelp.ithome.com.tw/upload/images/20241010/20121052nMHLoTTHrO.png

程式碼說明

db = client.customer_db
customer_collection = db.customers

1.** db = client.customer_db**
這行代碼是在選擇或創建一個名為 "customer_db" 的數據庫。
在 MongoDB 中,如果指定的數據庫不存在,它會在你首次存儲數據時自動創建
client 是之前創建的 MongoDB 客戶端連接。
.customer_db 指定了我們要使用的數據庫名稱。

2.customer_collection = db.customers
這行代碼是在選擇或創建一個名為 "customers" 的集合(collection)。
MongoDB 中,集合類(collection)似於關係型數據庫中的表(table)
如果 "customers" 集合不存在,MongoDB 會在你首次插入數據時自動創建它。
db.customers 指定了我們要使用的集合名稱。
總的來說,這兩行代碼的作用是:
選擇(或創建)一個名為 "customer_db" 的數據庫。
在該數據庫中選擇(或創建)一個名為 "customers" 的集合。
之後,我們就可以使用 customer_collection 來進行各種數據操作,如插入、查詢、更新和刪除顧客數據。這種方式允許我們直接與 MongoDB 數據庫進行交互,而不需要使用 JSON 文件來存儲數據。

# 創建顧客
@app.post("/customers/", response_model=CustomerInDB)
async def create_customer(customer: Customer):
    customer_dict = customer.dict()
    result = await customer_collection.insert_one(customer_dict)
    created_customer = await customer_collection.find_one({"_id": result.inserted_id})
    return CustomerInDB(id=str(created_customer["_id"]), **created_customer)
  1. customer_dict = customer.dict()
    將 Customer 對象轉換為字典,以便存入 MongoDB。

  2. insert_one() 方法用於插入單個文檔。

result = await customer_collection.insert_one(customer_dict)
異步插入顧客數據到 MongoDB。

  1. **created_customer = await customer_collection.find_one({"_id": result.inserted_id}) **

透過剛剛建立的resulID來把剛剛的物件回傳回來給前端看

有沒有覺得MongoDB挺簡單的XDD
可能比操作JSON還來的簡單,但是操作資料庫要非常小心
而且要避免掉一些error的情境發生喔~!!

其他的操作可以以此類推,幫大家整理在這

函數名稱 說明 使用範例
insert_one(document) 將一個文檔插入到集合中,返回插入結果。 result = await customer_collection.insert_one(customer_dict)
find_one(filter) 根據指定的過濾條件查找並返回第一個匹配的文檔。如果沒有匹配,則返回 None created_customer = await customer_collection.find_one({"_id": result.inserted_id})
update_one(filter, update) 根據過濾條件更新第一個匹配的文檔,並返回更新結果。 result = await customer_collection.update_one({"_id": ObjectId(customer_id)}, {"$set": updated_customer.dict()})
delete_one(filter) 根據過濾條件刪除第一個匹配的文檔,並返回刪除結果。 result = await customer_collection.delete_one({"_id": ObjectId(customer_id)})
find(filter) 根據指定的過濾條件查找所有匹配的文檔,返回一個游標。 customers = await customer_collection.find().to_list(1000)

說明
insert_one(document):用於將新客戶資料插入到 MongoDB 的集合中。
find_one(filter):用於查找剛插入的客戶資料,根據其 _id 獲取完整資料。
update_one(filter, update):用於更新特定客戶的資料,如果未找到該客戶則返回 404 錯誤。
delete_one(filter):用於刪除特定客戶資料,如果未找到該客戶則返回 404 錯誤。
find(filter):用於獲取所有顧客資料,並將其轉換為列表以便返回。

常用的MongoDB函數整理

操作類型 函數 說明
連接與數據庫操作 MongoClient() 連接到 MongoDB 數據庫實例。
client.database_name 訪問特定的數據庫(MongoDB 自動創建數據庫)。
db.collection_name 訪問特定集合。
插入操作 insert_one(document) 插入一條文件到集合中。
insert_many([doc1, doc2]) 插入多條文件到集合中。
inserted_id 獲取 insert_one() 後新插入文件的 _id
查詢操作 find_one(filter) 查詢集合中的第一條匹配的文件。
find(filter) 返回所有匹配條件的文件,返回一個游標(cursor)對象。
find({}, {"field": 1}) 返回所有文件,只顯示指定字段(projection)。
sort([("field", 1)]) 對查詢結果排序,1 為升序,-1 為降序。
limit(n) 限制查詢結果的數量。
count_documents(filter) 返回符合條件的文件數量。
更新操作 update_one(filter, {"$set": update}) 更新符合條件的第一條文件。
update_many(filter, {"$set": update}) 更新所有符合條件的文件。
replace_one(filter, new_document) 用新文件替換匹配的文件。
刪除操作 delete_one(filter) 刪除符合條件的第一條文件。
delete_many(filter) 刪除所有符合條件的文件。
集合操作 collection.drop() 刪除集合及其所有數據。
collection.rename(new_name) 重命名集合。
create_index([("field", 1)]) 為指定字段創建索引,1 為升序,-1 為降序。
list_indexes() 列出集合中的所有索引。
聚合操作 aggregate(pipeline) 執行聚合管道(pipeline),用於執行複雜的查詢或數據處理。
group 按字段分組,類似於 SQL 的 GROUP BY
match 用於篩選符合條件的文件,類似於 SQL 的 WHERE
project 用於選擇要返回的字段,類似於 SQL 的 SELECT
事務處理 client.start_session() 啟動一個事務會話。
session.start_transaction() 開始一個事務。
session.commit_transaction() 提交事務。
session.abort_transaction() 回滾事務。

總結

今天學會了幾個個重點

學習項目 重點 應用場景
API文件 - 清楚定義 API 路徑、參數與返回值 - 內部系統或外部系統之間的數據交換
- 使用標準如 OpenAPI 進行規範化 - 開發者和使用者快速理解 API 功能和調用方式
- 提供詳細的錯誤代碼與響應訊息 - 自動生成 API 文檔,減少文檔編寫的負擔
認證機制 - 使用 Cookie Session、OAuth、JWT 等方法進行身份驗證 - 需要對用戶進行登錄、身份驗證及權限管理的系統(如電商、社交網站)
- 確保安全性,如 Token 加密、過期時間設置 - 保護敏感數據和限制未授權用戶訪問
- 避免常見安全漏洞如 CSRF、XSS (未學到,但推薦去survery) - 多個微服務或系統之間的安全交互
資料庫應用 - 選擇合適的資料庫(如 SQL、NoSQL) - 電商平台需要高效查詢與數據一致性的應用場景
- 掌握 CRUD 操作、事務處理與索引優化(未學到,但推薦去survery) - 需要大規模數據存儲與訪問的應用(如社交網站、大數據平台)
- 保證資料庫的備份與恢復,並應對高併發(未學到,但推薦去survery) - 多租戶系統或用戶數據安全性要求高的應用場景

希望大家都收穫滿滿~
明天會開始進入PowerBI教學喔~!!


上一篇
[Day24] Python專案 - 網頁開發 - (3) 前端的至尊React與Python之間的交流
下一篇
[Day26] Python專案 - PowerBI - 視覺化你的報表(Excel、MongDB、Python結合)
系列文
從概念到應用:Python實戰開發學習之旅31
圖片
  直播研討會
圖片
{{ item.channelVendor }} {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言